/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.benchmark;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.client.DataPutter;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.datatype.PackableBytes;
import org.sss.mapreduce.datatype.PackableString;

public class IoCompositeInit {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(IoCompositeInit.class);
  public static class ValGenMapper extends Mapper {
    private byte[] realValue;
    
    @Override
    public void configure(Context context) throws Exception {
      Configuration conf = context.getConfiguration();
      int initialValueLength = Integer.parseInt(conf.get("initialValueLength"));
      realValue = new byte[initialValueLength];
      Arrays.fill(realValue, (byte)0);
    }

    public void map(Context context,
        PackableString key, PackableBytes value,
        Output<PackableString, PackableBytes> output) throws Exception {
      output.write(key, new PackableBytes(realValue));
    }
  }

  private static GroupID put(SssClient client, IoCompositeProperties prop, int valSize) throws Exception {
    byte[] v = new byte[valSize];
    Arrays.fill(v, (byte)0);

    DataPutter<PackableString, PackableBytes> putter = 
      DataPutter.create(client, PackableString.class, PackableBytes.class);
    GroupID table = putter.getOutputGroupID();
    try {
      for (int i = 0; i < prop.getInitialKeyCount(); i++) {
        String k = String.format("input-%016x", i);
        putter.append(new PackableString(k), new PackableBytes(v));
      }
    } finally {
      putter.close();
    }
    return table;
  }

  /**
   * Usage:
   * java IoCompositeInit
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);
    String[] others = client.getArgs();
    GroupID output = null;

    IoCompositeProperties prop = new IoCompositeProperties(others);

    int size = prop.getInitialValueLength();
    if (size > 16 * 1024) { // 16kB limit
      GroupID table = put(client, prop, 1);

      JobEngine engine = new JobEngine(client);
      try {
        output = GroupID.createRandom(engine);

        Properties pr = prop.getProperties();
        for (Map.Entry<Object, Object> e: pr.entrySet()) {
          String key = (String)e.getKey();
          String value = (String)e.getValue();
          engine.getConfiguration().set(key, value);
        }
        engine.getJobBuilder("Gen", ValGenMapper.class)
          .addInput(table).addOutput(output).build();
        engine.exec();
      } finally {
        engine.dispose();
      }
    }
    else {
      output = put(client, prop, size);
    }
    System.out.println("output data deployed - " + output.toString());
  }
}
